package org.example

import org.apache.spark.sql.SparkSession

/**
 * Spark读取Flink创建的Hudi表示例 - 多表读取版本
 * 需要Hudi Spark运行时依赖：org.apache.hudi:hudi-spark3.2-bundle_2.12
 */
object SparkHudiMultiTableReader {
    def main(args: Array[String]): Unit = {
        // 初始化SparkSession
        val spark = SparkSession.builder()
            .appName("Hudi Multi Table Spark Reader")
            .master("local[1]")  // 本地模式运行
            .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
            .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
            .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
            .config("spark.sql.hive.convertMetastoreParquet", "false")
            .getOrCreate()

        // 设置日志级别为WARN以减少日志输出
        spark.sparkContext.setLogLevel("WARN")
        
        // 定义表路径
        val userTablePath = "/tmp/hudi/user_info_mor"
        val orderTablePath = "/tmp/hudi/order_info_mor"

        println("将用户信息Hudi表注册为临时表:")
        spark.sql(
            s"""
            |CREATE TEMPORARY VIEW user_info_view
            |USING hudi
            |OPTIONS (
            |  path '$userTablePath'
            |)
            |""".stripMargin)
        
        println("将订单信息Hudi表注册为临时表:")
        spark.sql(
            s"""
            |CREATE TEMPORARY VIEW order_info_view
            |USING hudi
            |OPTIONS (
            |  path '$orderTablePath'
            |)
            |""".stripMargin)
        
        println("\n=== 用户信息表数据 ===")
        spark.sql("SELECT * FROM user_info_view").show(false)
        
        println("\n=== 订单信息表数据 ===")
        spark.sql("SELECT * FROM order_info_view").show(false)
        
        println("\n=== 用户信息表Schema ===")
        spark.sql("SELECT * FROM user_info_view").printSchema()
        
        println("\n=== 订单信息表Schema ===")
        spark.sql("SELECT * FROM order_info_view").printSchema()
        
        // 使用DataFrame API读取Hudi表
        println("\n=== 使用DataFrame API读取用户信息表 ===")
        val userDF = spark.read.format("hudi").load(userTablePath)
        userDF.show(false)
        
        println("\n=== 使用DataFrame API读取订单信息表 ===")
        val orderDF = spark.read.format("hudi").load(orderTablePath)
        orderDF.show(false)
    }
} 